#include "event.h"

#include <sys/epoll.h>
#include <unistd.h>
#include <time.h>
#include <cassert>
#include <cstring>

#include "timer.h"
#include "common/log.h"


namespace evm { namespace event {

static long timespec_to_ms(const timespec& ts) {
    return static_cast<long>(ts.tv_sec * 1e3 + ts.tv_nsec / 1e6);
}

EventLoop::EventLoop(unsigned int maxevents)  {
    epoll_fd_ = epoll_create(1);
    assert(epoll_fd_ != -1);
    assert(maxevents > 0);
    epoll_events_ = new epoll_event[maxevents];
    epoll_maxevents_ = maxevents;
}

EventLoop::~EventLoop() {
    if (running_)
        stop();
    close(epoll_fd_);
    delete [] epoll_events_;
}

bool EventLoop::add_event(int fd, int mask, FileHandler handler) {
    lock_guard<mutex> lg(mutex_);

    if (mask == EV_NONE)
        return false;

    auto& fe = file_events_[fd];

    /* If the fd was already monitored for some event, we need a MOD
     * operation. Otherwise we need an ADD operation. */
    int op = (fe.mask == EV_NONE) ? EPOLL_CTL_ADD : EPOLL_CTL_MOD;

    epoll_event ee;
    ee.events = EPOLLET;  // edge trigger
    ee.data.u64 = 0;  // avoid valgrind warning
    ee.data.fd = fd;

    fe.mask |= mask;  // merge old events
    if (fe.mask & EV_READABLE)
        ee.events |= EPOLLIN;
    if (fe.mask & EV_WRITABLE)
        ee.events |= EPOLLOUT;

    if (epoll_ctl(epoll_fd_, op, fd, &ee) < 0)
        return false;

    fe.same_rw = mask & EV_READABLE && mask & EV_WRITABLE;

    if (mask & EV_READABLE)
        fe.rhandler = handler;
    if (mask & EV_WRITABLE)
        fe.whandler = handler;

    return true;
}

bool EventLoop::rmv_event(int fd, int mask) {
    lock_guard<mutex> lg(mutex_);

    if (!file_events_.count(fd))
        return true;

    auto& fe = file_events_[fd];
    // if (fe.mask == EV_NONE)
    //     return true;

    fe.mask &= ~mask;

    epoll_event ee;
    ee.events = 0;
    ee.data.u64 = 0;  // avoid valgrind warning
    ee.data.fd = fd;

    if (fe.mask & EV_READABLE)
        ee.events |= EPOLLIN;
    if (fe.mask & EV_WRITABLE)
        ee.events |= EPOLLOUT;
    int op = (fe.mask == EV_NONE) ? EPOLL_CTL_DEL : EPOLL_CTL_MOD;

    /* Note, Kernel < 2.6.9 requires a non null event pointer even for
     * EPOLL_CTL_DEL. */
    if (epoll_ctl(epoll_fd_, op, fd, &ee) < 0)
        return false;

    // Delete event no longer polled.
    if (fe.mask == EV_NONE)
        file_events_.erase(fd);

    return true;
}

size_t EventLoop::add_timer(double seconds, TimeHandler handler) {
    log_info("add_timer, {}", seconds);
    lock_guard<mutex> lg(mutex_);

    assert(seconds >= 0);

    auto id = ++next_time_id_;

    timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) < 0)
        return EV_INVALID_ID;

    long when = timespec_to_ms(ts) + static_cast<long>(seconds * 1e3);

    /* When computed 'when' is duplicate, offset it by 1 ms. */
    while (time_events_.count(when))
        ++when;

    auto& te = time_events_[when];
    te.id = id;
    te.when = when;
    te.handler = handler;
    te.interval = seconds;

    return id;
}

bool EventLoop::rmv_timer(size_t id) {
    lock_guard<mutex> lg(mutex_);

    if (id == EV_INVALID_ID)
        return false;

    auto iter = time_events_.begin();
    for (; iter != time_events_.end(); ++iter) {
        if (iter->second.id == id)
            break;
    }
    if (iter == time_events_.end())
        return false;

    time_events_.erase(iter);
    return true;
}

// Run event loop forever, until being stopped.
void EventLoop::run() {
    if (running_)
        return;

    running_ = true;
    while (running_) {
        /*LOG->info() << "loop run";*/
        process_file_events();
        process_time_events();
    }
}

void EventLoop::start() {
    if (running_)
        return;

    thread_ = thread([this]() {
        run();
    });

    log_info("event loop started");
}

void EventLoop::stop(bool wait) {
    if (!running_)
        return;

    running_ = false;

    /* Wake up the epoll wait, within 1ns. */
    OneshotTimer oneshot(1e-9, [] {
        log_info("epoll wait woken up");
    }, this);

    if (wait) {
        if (thread_.joinable())
            thread_.join();
    }
    log_info("event loop stopped");
}

void EventLoop::process_file_events() {
    int msec = -1;  // wait forever if no time or file event arrives

    {
        lock_guard<mutex> lg(mutex_);

        // cppcheck-suppress stlSize
        if (time_events_.size() > 0) {
            auto te = time_events_.begin();

            timespec now_ts;
            if (clock_gettime(CLOCK_MONOTONIC, &now_ts) < 0)
                return;
            long now = timespec_to_ms(now_ts);
            long when = te->first;

            if (when < now) {
                msec = 0;  // time event arrived, so no wait
            }
            else {
                msec = static_cast<int>(when - now);
            }

            log_info("process_file_events, now {}, when {}, msec {}, this {}",
                     now, when, msec, reinterpret_cast<uintptr_t>(this));
        }
    }

    int nfds = epoll_wait(epoll_fd_, epoll_events_, epoll_maxevents_, msec);
    if (nfds < 0) {
        log_error("epoll_wait error: {}, {}", errno, strerror(errno));
    }
    else if (nfds > 0) {
        for (int i = 0; i < nfds; ++i) {
            int mask = 0;
            auto e = epoll_events_ + i;
            if (e->events & EPOLLIN)
                mask |= EV_READABLE;
            if (e->events & EPOLLOUT || e->events & EPOLLERR
                || e->events & EPOLLHUP)
                mask |= EV_WRITABLE;
            int fd = e->data.fd;

            FileHandler rhandler;
            FileHandler whandler;

            {
                 lock_guard<mutex> lg(mutex_);

                /* Maybe an already processed event removed an element
                 * that fired and we still didn't processed,
                 * so we check if the event is still valid. */
                if (!file_events_.count(fd))
                    continue;

                const auto &fe = file_events_[fd];

                bool rfired = false;

                if (fe.mask & mask & EV_READABLE) {
                    rhandler = fe.rhandler;
                    rfired = true;
                }

                if (fe.mask & mask & EV_WRITABLE && !(rfired && fe.same_rw))
                    whandler = fe.whandler;
            }

            if (rhandler)
                try {
                    rhandler(fd, mask);
                } catch (const exception& e) {
                    log_error("file event: {}", e.what());
                }

            {
                lock_guard<mutex> lg(mutex_);
                if (!file_events_.count(fd))
                    continue;
            }

            if (whandler)
                try {
                    whandler(fd, mask);
                } catch (const exception& e) {
                    log_error("file event: {}", e.what());
                }
        }
    }

    log_info("process_file_events");
}

void EventLoop::process_time_events() {
    unique_lock<mutex> lock(mutex_);

    for (auto iter = time_events_.begin();
         iter != time_events_.end();
         iter = time_events_.begin())  // obtain begin iter again every time
    {
        timespec now_ts;
        if (clock_gettime(CLOCK_MONOTONIC, &now_ts) < 0)
            return;
        long now = timespec_to_ms(now_ts);

        long when = iter->first;

        // log_info("process_time_events, now {}, when {}, this {}", now, when, reinterpret_cast<uintptr_t>(this));
        if (now < when)
            break;  // no more time events need to be processed now

        auto& e = iter->second;
        double interval;
        lock.unlock();
        try {
            interval = e.handler(e.id);  // handle time event
        } catch (const exception& e) {
            log_error("{}, {}", reinterpret_cast<uintptr_t>(this), e.what());
            interval = EV_PERIODIC;
        }
        lock.lock();

        if (interval == EV_NOMORE) {
            time_events_.erase(iter);  // no next time event, remove it
            continue;
        }
        else if (interval < 0)
            throw invalid_argument("invalid time event interval");
        else if (abs(interval - EV_PERIODIC) < 1e-3)
            interval = e.interval;  // use last time interval
        else
            e.interval = interval;  // update next event time interval

        timespec when_ts;
        if (clock_gettime(CLOCK_MONOTONIC, &when_ts) < 0)
            return;
        e.when = timespec_to_ms(when_ts) + static_cast<long>(interval * 1e3);

        /* When computed 'when' is duplicate, offset it by 1 ms. */
        while (time_events_.count(e.when))
            ++e.when;

        log_info("process_time_events, when {}, interval {}, id {}, this {}",
                 e.when, interval, e.id, reinterpret_cast<uintptr_t>(this));
        time_events_[e.when] = e;  // add updated time event
        time_events_.erase(iter);  // remove outdated time event
    }
}

/*long time() {
    timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) < 0) {
        assert(errno == EINVAL);
        throw invalid_argument("CLOCK_MONOTONIC is not supported");
    }
    return timespec_to_ms(ts);
}*/

}}

